package cn.doitedu.demo1;


import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/7
 * @Desc: 学大数据，上多易教育
 *
 *   按如下规则，进行计算：
 *      用户发生 A行为：3次以上
 *      用户发生  [ Q C Y ] 事件序列:  2次以上
 *
 *      当他做 C行为时，触发规则
 *
 **/
public class Exercise1 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));

        // 构建kafka source来读取数据
        KafkaSource<String> source = KafkaSource
                .<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("user-action-log")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("doitedu-ww")
                .setClientIdPrefix("doitedu-cc")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "log-source");

        // 将source读取到的json数据，转换成javaBean数据
        DataStream<UserAction> mapped = stream.map(json -> JSON.parseObject(json, UserAction.class));


        // 按用户分区
        KeyedStream<UserAction, Long> keyedStream = mapped.keyBy(UserAction::getUser_id);


        keyedStream.process(new KeyedProcessFunction<Long, UserAction, String>() {

            ValueState<Integer> ACNT;
            ValueState<Integer> IDX;
            ValueState<Integer> SEQ_CNT;
            final String[] eventSeq = {"Q","C","Y"};
            JSONObject message;

            @Override
            public void open(Configuration parameters) throws Exception {

                RuntimeContext runtimeContext = getRuntimeContext();


                ACNT = runtimeContext.getState(new ValueStateDescriptor<Integer>("A_CNT", Integer.class));
                IDX = runtimeContext.getState(new ValueStateDescriptor<Integer>("IDX", Integer.class));
                SEQ_CNT = runtimeContext.getState(new ValueStateDescriptor<Integer>("SEQ_CNT", Integer.class));

                message = new JSONObject();
                message.put("rule_id","rule-002");

            }

            @Override
            public void processElement(UserAction userAction, KeyedProcessFunction<Long, UserAction, String>.Context ctx, Collector<String> out) throws Exception {

                String eventId = userAction.getEvent_id();

                /* *
                 * 用户发生 A行为：3次以上
                 * 用户发生  [ Q C Y ] 事件序列:  2次以上
                 * 当他做 C行为时，触发规则
                 */

                // A行为统计
                if(eventId.equals("A")){
                    int cnt = ACNT.value() == null? 1: ACNT.value()+1;
                    ACNT.update(cnt);
                }

                // [ Q C Y ] 事件序列的发生次数统计
                int idx = IDX.value() == null ? 0: IDX.value();
                if(eventId.equals( eventSeq[idx]  )){
                    // 如果是，则要把idx 递增
                    int newIdx = idx+1;
                    // 如果idx递增后等于3了，说明这个序列到此被完整发生过了
                    if(newIdx == 3){
                        // 重置idx
                        newIdx = 0;
                        // 增加完整发生次数
                        SEQ_CNT.update( SEQ_CNT.value() == null? 1 : SEQ_CNT.value()+1  );
                    }
                    IDX.update(newIdx);
                }

                // 触发事件C行为 处理
                if(eventId.equals("C")){
                    Integer aCount = ACNT.value();
                    Integer seqCount = SEQ_CNT.value();

                    if(aCount != null && seqCount != null && aCount>=3 && seqCount>=2){

                        message.put("urser_id",userAction.getUser_id());
                        message.put("hit_time",userAction.getAction_time());

                        out.collect(message.toJSONString());

                    }
                }
            }
        }).print();


        env.execute();


    }
}
